package com.ku.rocketmq.quickstart.master_slave;

import com.ku.rocketmq.constants.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * @program: rocketmqDemo
 * @description: 主从模式 消息消费者
 * @author: JokerKu
 * @create: 2020-05-03 13:53
 **/
public class Consumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_consumer_name");

        //1.设置nameserver 地址
        consumer.setNamesrvAddr(Const.NAMESERVER_ADDR);
        //2.设置消费者消费消息的起点 这里从最后面开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //3.订阅topic
        consumer.subscribe("test_quick_topic","TagA");


        //rocketMQ Consumer 核心配置项
//        consumer.setConsumeMessageBatchMaxSize(10);//一次最大能拉取消息的数量
//        consumer.setConsumeThreadMax(20);//消费者并行线程的最大数量
//        consumer.setConsumeThreadMin(10);//消费者并行线程的最小数量
//        consumer.setConsumeTimeout(1000);//消费超时的时间
//        consumer.setMaxReconsumeTimes(3);//消息消费失败的重试次数
//        consumer.setMessageModel(MessageModel.CLUSTERING);//设置为集群模式
//        consumer.setMessageModel(MessageModel.BROADCASTING);//设置为广播模式

        //4.监听该topic
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt message = msgs.get(0);
                try {
                    String topic = message.getTopic();
                    String tags = message.getTags();
                    String keys = message.getKeys();

                    //模拟出错
                    if (keys.equals("keyA3")){
                        int i = 3/0;
                    }
                    String msgBody = new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    System.out.println("topic:"+topic+"--tags:"+tags+"--keys:"+keys+"--message:"+msgBody);

                }catch (Exception e){
                    e.printStackTrace();
                    //如果出错...这里可以进行日志记录
                    int reconsumeTimes = message.getReconsumeTimes();
                    System.err.println("reconsumeTimes:"+reconsumeTimes);
                    if (reconsumeTimes==3){
                        //不让消息一直重发
                        //在这里做日志和补偿处理
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    //发生异常 返回重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                //消息消费成功，返回success
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            }
        });

        //别忘了最后一定要启动consumer
        consumer.start();
        System.out.println("MQConsumer..start...");
    }
}
